package com.flink.split_stream.connect;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * 描述:
 * 合流-连接
 * TODO 合流连接方式是将两个不同类型的流通过连接的方式保存原始数据的类型以便于进行编写清洗逻辑，最后合并成一个同类型的流输出
 *
 * @author yanzhengwu
 * @create 2022-08-15 0:37
 */
public class ConnectStream {


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Integer> stream1 = env.fromElements(1, 2, 3);
        DataStreamSource<Long> stream2 = env.fromElements(4L, 5L, 6L);

        //TODO 重点： 返回的 ConnectedStreams IN1 是调用者 IN2是被调用者  谁调用connect方法谁是IN1
        stream1.connect(stream2)
                .map(new CoMapFunction<Integer, Long, String>() {
                    @Override
                    public String map1(Integer value) throws Exception {
                        return "Integer:" + value;
                    }

                    @Override
                    public String map2(Long value) throws Exception {
                        return "Long:" + value;
                    }
                }).print();


        env.execute();
    }
}
